/*
 * 	prmMQlib.h
 *
 *  Created on: Feb 13, 2012
 *  Author: Sangcheol Charles MIN
 */
#ifndef __PRMMQLIB_H_
#define __PRMMQLIB_H_

#include <decaf/lang/Thread.h>
#include <decaf/lang/Runnable.h>
#include <decaf/util/concurrent/CountDownLatch.h>
#include <decaf/lang/Long.h>
#include <decaf/lang/Integer.h>
#include <decaf/util/Date.h>
#include <decaf/util/concurrent/ConcurrentStlMap.h>
#include <activemq/core/ActiveMQConnectionFactory.h>
#include <activemq/core/ActiveMQConnection.h>
#include <activemq/transport/DefaultTransportListener.h>
#include <activemq/util/Config.h>
#include <activemq/library/ActiveMQCPP.h>
#include <cms/Connection.h>
#include <cms/Session.h>
#include <cms/TextMessage.h>
#include <cms/BytesMessage.h>
#include <cms/MapMessage.h>
#include <cms/ExceptionListener.h>
#include <cms/MessageListener.h>
#include <stdlib.h>
#include <stdio.h>
#include <memory>
#include <iostream>
#include <map>
#include <Poco/Timer.h>

#ifdef __cplusplus
extern "C" {
#endif

#define PRMMQLIB_HEADER_VERSION "$Revision: 1.00 $"

//ERROR Message

using namespace activemq;
using namespace activemq::core;
using namespace activemq::transport;
using namespace decaf;
using namespace decaf::lang;
using namespace decaf::util;
using namespace decaf::util::concurrent;
using namespace cms;
using namespace std;

namespace prmMQlib
{

typedef enum {
	PRMMQLIB_API_SUCCESS = 				0,		// Value when PRMMQLIB API is successfully done
	PRMMQLIB_WARNINIG = 				-1,		// Value when PRMMQLIB API has warning
	PRMMQLIB_API_ERR_UNKNOWN =			-2,		// Value when PRMMQLIB API has error but it is not handled in the library
	PRMMQLIB_API_ERR_INVALID = 			-3,		// Value when PRMMQLIB API status is not normal status
	PRMMQLIB_API_ERR_INVALID_PARAMS	=	-4		// Value when the parameters API is using are not appropriate
} PrmGeneralReturnCode ;

/*
 *  If user wants his application to be a consumer,
 *  one of his classes must inherit ConsumerClient to get a message callback function.
 *  Through rebuilding 'OnMessage()", client application can get messages from producer
 */
class ConsumerClient
{
public:
	virtual void OnMessage(int mesgType, char* mesg, int length) const 	= 0;
	/* Description 	: Message Receiver for consumer
	 * mesgType 	: TXT_MESSAGE_TYPE or BIN_MESSAGE_TYPE
	 * mesg 		: text and binary has 'char*'. User should modify it when byte mesg is handled
	 */

    virtual void onException( const CMSException& ex AMQCPP_UNUSED ) 	= 0;
    /* Exceptional handler for communication between producer and consumer
     * User can get
     *         virtual std::string getMessage() const;
     *         virtual const std::exception* getCause() const;
     *         virtual std::vector< std::pair< std::string, int> > getStackTrace() const;
     *         virtual void printStackTrace() const;
     *         virtual void printStackTrace( std::ostream& stream ) const;
     *         virtual std::string getStackTraceString() const;
     *         virtual const char* what() const throw(); // const char pointer to error message
     */

    virtual void transportInterrupted() = 0;			// called when network transportation is broken or breached
    virtual void transportResumed() 	= 0;			// called when network transportation is resumed.
    virtual void OnMessageRequest(int mesgType , char* buff, int length,const Destination*, string mesgId) {};
};



/////////////////////////////////////////////////////////////////////////////////////////////////
// THIS IS MAIN PART. PLEASE GO TO "USER INTERFACE PART"
class CPrmMQlib
{
#define TXT_MESSAGE_TYPE		0x01		// text message type : used in OnMessage
#define BIN_MESSAGE_TYPE		0x02

// Definition of inner classes
private :
	class CProducer {
	private :
		Connection* mqConnection;
		Destination* destination;
		bool useTopic;
		bool clientAck;
		string topicQueueName;

	private :
		CProducer( const CProducer& );
		CProducer& operator= ( const CProducer& );

	public:
		Session* session;
		MessageProducer* producer;
	public :
		CProducer(Connection* , const std::string&, bool,bool);

		~CProducer();
		int sendMessage(string& textContent);
		int sendMessage(unsigned char* ByteContent,int& );
	};


	class CRequestProducer : 	public CProducer,
								public MessageListener {
	private:
		Destination* 		tempDest;
		MessageConsumer* 	responseConsumer;
		ConsumerClient*		mesgCallback;

	private :
		CRequestProducer( const CRequestProducer& );
		CRequestProducer& operator= ( const CRequestProducer& );

	public :
		CRequestProducer(	Connection* connection,
							const std::string& topicName,
							bool isTopic,
							bool clientAck,
							ConsumerClient* ); // : CProducer(connection, topicName, isTopic, clientAck) {}

		~CRequestProducer( void	) throw();

	    virtual void onMessage( const cms::Message*) throw();
		int sendMessage(string messageId, string& textContent);
		int sendMessage(string messageId, unsigned char* ByteContent,int& );
	};

	class CConsumer : 	public ExceptionListener,
						public MessageListener,
						public DefaultTransportListener {
	private :
		Connection* 		mqConnection;
		Session* 			session;
		Destination* 		destination;
	    MessageConsumer* 	consumer;
	    MessageProducer* 	replyProducer;
	    ConsumerClient*		mesgCallback;
		bool useTopic;
		bool clientAck;
		string topicQueueName;
		string userID;

	private:
		CConsumer( const CConsumer& );
		CConsumer& operator= ( const CConsumer& );

	public:
		bool isTemperery;

	public:
	 	CConsumer(	Connection*, const string&, ConsumerClient*, bool, bool);
	 	CConsumer(	Connection*, const string&, ConsumerClient*, const string&);
	 	CConsumer(	Connection*, const string&, ConsumerClient*, bool, bool, bool);
		virtual ~CConsumer() throw();
	    virtual void onMessage( const cms::Message*) throw();
	    virtual void onException( const CMSException& ex AMQCPP_UNUSED );
	    virtual void transportInterrupted();
	    virtual void transportResumed();

	    int replyMessage(	const Destination* replyTo,		/* request destination such as client 												*/
	    					const string messageId,			/* should be correlationId of request message. cf) message->getCMSCorrelationID()	*/
	    					string& strContent
	    				);

	    int replyMessage(	const Destination* replyTo,
	    					const string messageId,
	    					unsigned char* buffer,	/* Byte based message - it will be google proto buf message */
	    					int& len				/* the length of message sent 								*/
	    				);

	};

protected :
	Connection* mqConnection;				/* it indicates mq connection contents 	*/
	string brokerURI;						/* this is well-known broker URI		*/
	string failoverBrokerURI;				/* it would be used later 				*/
	ConcurrentStlMap<string, CProducer*> producers;
	ConcurrentStlMap<string, CConsumer*> consumers;
	vector<string> producer_topics;
	vector<string> consumer_topics;

public:
	/* Logger_t logger*/

//////////////////////////////////////////////////////////////////////////////////////////////////////////////////
// USER INTERFACE PART
private:
	CPrmMQlib( const CPrmMQlib& );
	CPrmMQlib& operator= ( const CPrmMQlib& );
	string getTopicNameFromServiceId( const string& );

// Definition of active mq interfaces
public:
	CPrmMQlib();

	virtual ~CPrmMQlib();

	PrmGeneralReturnCode initPrmMQlib(std::string& brokerURI);	/* 	initialize active MQ module,
	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 *	connect well-known active MQ */

	PrmGeneralReturnCode initPrmMQlib(std::string& brokerURI,
									const string& clientID	);	/* 	initialize active MQ module,
	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 *	connect well-known active MQ with clientID */

	PrmGeneralReturnCode addProducer(
					const string& TopicQueueName,	/* Add producer information into producers map 	*/
					const bool useTopic = false,	/* create topic and setup active mq producer	*/
					const bool clientAck = false,
					ConsumerClient* client = NULL
					);

	PrmGeneralReturnCode addProducerfromServiceId(
					const string& serviceId
	);


	PrmGeneralReturnCode addConsumer(
					ConsumerClient* client,			/* The class calling MesgCallback				*/
					const string& TopicQueueName,	/* Consumer topic/queue Name					*/
					bool useTopic = false,			/* if topic, set to 'true', otherwise queue		*/
					bool clientAck = false			/* does it need courterpart's mq reply? then true*/
					);

	PrmGeneralReturnCode addConsumer(
					ConsumerClient* client,
					const string& TopicOrQueueName,
					bool useTopic,
					bool clientAck,
					bool isTemperery
					);

	PrmGeneralReturnCode addDurableConsumer(
					ConsumerClient* client,			/* The class calling MesgCallback				*/
					const string& TopicQueueName,	/* Consumer topic/queue Name					*/
					const string& applicationName
					);

	int sendTextMessage(
			const std::string& TopicOrQueueName,
			std::string& buffer
			);

	int sendBytesMessage(
			const std::string& TopicOrQueueName,
			unsigned char* buffer,
			int& length,
			std::string correlation_mesg_id = ""
			);

	int replyTextMessage (
			const Destination* replyTo,
			const string message_id,
			const std::string& TopicOrQueueName,
			std::string& textMessage
			);

	int replyBytesMessage(
			const Destination* replyTo,
			const string message_id,
			const std::string& TopicOrQueueName,
			unsigned char* buffer,
			int& length
			);

	vector<string> getRegisteredTopicName(bool isProducer) const{
		if(isProducer == true )
			return producer_topics;
		else
			return consumer_topics;
	}

	virtual void onException( std::string& ex ) {
		exit(1);
	}


};

}


#ifdef __cplusplus
}
#endif
#endif /* __PRMMQLIB_H_ */
